/*
    Copyright 2010 Northbranchlogic, Inc.

    Licensed under the Apache License, Version 2.0 (the "License");
    you may not use this file except in compliance with the License.
    You may obtain a copy of the License at

        http://www.apache.org/licenses/LICENSE-2.0

    Unless required by applicable law or agreed to in writing, software
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.
 */

package com.norbl.runjrun;

import com.norbl.util.*;
import com.norbl.util.ec2.*;
import com.norbl.util.s3.*;
import com.norbl.util.sqs.*;
import com.norbl.util.sdb.*;
import com.xerox.amazonws.ec2.EC2Exception;
import com.xerox.amazonws.sdb.SDBException;
import com.xerox.amazonws.sqs2.*;
import org.jets3t.service.S3ServiceException;
import java.util.*;

/** This class contains all the operations that can be performed
 *  on the nodes and tasks.  While this class can be instantiated,
 *  it is primarily intended to be the the base for othere classes that
 *  perform operations
 *  ({@link Client}, {@link com.norbl.runjrun.gui.Actions}).<p>
 *
 *  <b><i>Note</i></b> that when using this class, you <i>must </i> call
 *  its {@link #init(com.norbl.util.ec2.Ec2Ops, com.norbl.util.s3.S3Ops, com.norbl.util.sdb.SdbOps, com.norbl.util.sqs.SqsOps, com.norbl.runjrun.NodeStateTable, com.norbl.runjrun.TaskStateTable, java.lang.String, java.lang.String) }
 *  method  before using any of it's other methods.
 *
 * @author Barnet Wagman
 */
public class NodeTaskOps {

    private Ec2Ops eops;
    private S3Ops sops;
    private SdbOps dops;
    private SqsOps qops;

    public NodeStateTable nodeStateTable;
    public TaskStateTable taskStateTable;
    protected String inputBucketName;
    protected String inputS3KeyPrefix;

    public void init(Ec2Ops eops,S3Ops sops, SdbOps dops,SqsOps qops,
                     NodeStateTable nodeStateTable,
                     TaskStateTable taskStateTable,
                     String inputBucketName,
                     String inputS3KeyPrefix) {
        this.eops = eops;
        this.sops = sops;
        this.dops = dops;
        this.qops = qops;
        this.nodeStateTable = nodeStateTable;
        this.taskStateTable = taskStateTable;
        this.inputBucketName = inputBucketName;
        this.inputS3KeyPrefix = inputS3KeyPrefix;
    }

        //  -------- Node ops ---------------------------

        /** Removes all nodes from the node state table.
         *  This op has no effect on the actual {@link RunnerNode}
         *  instances or the hosts they are running on.  E.g. it does not
         *  terminate the ec2 instances.<p>
         *
         *  Note that a running {@link RunnerNode} will reappear
         *  after being cleared, since {@link RunnerNode} periodically
         *  sends 'pulse' messages to the sdb node table.
         */
    public void clearAllNodes() throws SDBException {       
        synchronized(nodeStateTable) {
            nodeStateTable.clear();
        }
    }

        /** Purges the node state table of nodes whose ec2 instances
         *  are no longer running (i.e. have state 'shutting-down' or
         *  'terminated'.  This operation does not effect
         *  node state entries for non-ec2 hosts.
         *
         */
    public void clearDeadInstances() throws SDBException {       
        synchronized(nodeStateTable) {           
            NEXT: for (NodeState s : nodeStateTable.currentStates ) {               
                if ( !s.isOnEc2Instance() ) continue NEXT;
                if ( NodeState.InstanceState.shutting_down.equals(s.instanceState) ||
                     NodeState.InstanceState.terminated.equals(s.instanceState) ||
                     NodeState.InstanceState.nil.equals(s.instanceState)) {
                    nodeStateTable.deleteItem(s.host,true);                  
                }
            }
        }
    }

        /** Terminates all ec2 instances that are listed in
         *  the node table (i.e. that are running
         *  RunJRun RunnerNodes).
         *
         */
    public void terminateInstances() throws EC2Exception {

        synchronized(nodeStateTable) {
            List<String> iids = new ArrayList<String>();
            NEXT: for (NodeState s : nodeStateTable.currentStates ) {
                if ( NodeState.InstanceState.pending.equals(s.instanceState) ||
                     NodeState.InstanceState.running.equals(s.instanceState) ) {
                    iids.add(s.instanceID);
                }
            }
                
            eops.terminateInstances(iids);
        }
    }

        /** Reboot instances with non-functioning nodes
         *  Reboot all instances whose RunnerNodes
         *  appear to have died - i.e. nodes with state
         * 'no_pulse' or 'nil'.
         *
         */
    public void rebootDefunctNodeInstances() throws EC2Exception {
        synchronized(nodeStateTable) {
            List<String> iids = new ArrayList<String>();
            NEXT: for (NodeState s : nodeStateTable.currentStates ) {
                if ( s.isOnEc2Instance()
                     &&
                     ( NodeState.Status.no_pulse.equals(s.status) ||
                       NodeState.Status.nil.equals(s.status) )
                     &&
                     NodeState.InstanceState.running.equals(s.instanceState) ) {
                    iids.add(s.instanceID);
                }
            }
            eops.rebootInstances(iids);
        }
    }


        /** Purges the node state table of nodes that are no longer
         *  sending pulses (and are presumably dead).  Note that this
         *  method will purge nodes whose instances are still running.
         *  This operation is rarely appropriate - in most cases
         *  {@link #clearDeadInstances() } is preferable.
         */
    public void clearNonPulsingNodes() throws SDBException {
        synchronized(nodeStateTable) {
            NEXT: for (NodeState s : nodeStateTable.currentStates ) {
                if ( NodeState.Status.no_pulse.equals(s.status) ) {
                    nodeStateTable.deleteItem(s.host,true);
                }
            }
        }
    }

        /** Removes the specified node from the node state table. This op
         *  has no effect on the actual {@link RunnerNode}
         *  instance or the host it is running on.  E.g. it does not
         *  terminate an ec2 instance.<p>
         *
         *  Note that a running {@link RunnerNode} will reappear
         *  after being cleared, since {@link RunnerNode} periodically
         *  sends 'pulse' messages to the sdb node table.
         *
         * @param hostName
         */
    public void clearNode(String hostName) throws SDBException {
        synchronized(nodeStateTable) {
            NodeState s = nodeStateTable.getNode(hostName);
            if ( s != null ) {
                nodeStateTable.deleteItem(hostName, true);
            }
        }
    }

        /** Reboots an ec2 instance. If the node is running a task
         *  when the reboot occurs, any work on that task will be
         *  lost.  The task will get requeued if auto-requeuing is
         *  enabled.
         *
         * @param hostName
         */
    public void rebootInstance(String hostName) {
        throw new UnsupportedOperationException();
    }

        // ------------ Task ops ------------------------------

        /** Deletes all task message from the sqs queue, all
         *  raw task objects from S3, and all task states from
         *  the sdb task table.  Note that a running task is not
         *  removed from it's node.  Cooked tasks in S3 are not affected.         
         */
    public void clearAllTasksEverywhere()
        throws SDBException, SQSException, S3ServiceException {
        synchronized(taskStateTable) {
            qops.deleteAllMessages();
            taskStateTable.deleteAllItems(true);
            sops.deleteObjects(inputBucketName,
                               inputS3KeyPrefix,
                               true);
        }
    }

        /** Removes all finished tasks from the task state table.        
         */
    public void clearFinishedTasks() throws SDBException {
        synchronized(taskStateTable) {
            for ( TaskState s : taskStateTable.currentStates ) {
                if ( TaskState.Status.finished.equals(s.status) ) {
                    taskStateTable.deleteItem(s.taskName, true);
                }
            }
        }
    }

    public void deleteWaitingTasks() throws SDBException, SQSException {
        synchronized(taskStateTable) {
            for ( TaskState s : taskStateTable.currentStates ) {
                if ( TaskState.Status.waiting.equals(s.status) ) {
                    Message m = getMessage(s);
                    if ( m != null ) qops.deleteMessage(m);
                    else System.out.println("No SQS task message was found in the queue " +
                        " for task=" + s.taskName);
                    taskStateTable.deleteItem(s.taskName, true);
                }
            }
        }
    }

    private Message getMessage(TaskState s) throws SQSException {

        if ( s.taskName == null ) throw new RunJRunException("Task name == null");

        List<Message> messages = qops.getAllMessages();
        for ( Message m : messages ) {
            TaskDescription td = new TaskDescription(m.getMessageBody());
            if ( s.taskName.equals(td.inputS3Key) ) return(m);
        }
        return(null);
    }

        /** Removes all orphaned tasks from the task state table.        
         */
    public void clearOrphanedTasks() throws SDBException {
        synchronized(taskStateTable) {
            for ( TaskState s : taskStateTable.currentStates ) {
                if ( TaskState.Status.orphan.equals(s.status) ) {
                    taskStateTable.deleteItem(s.taskName, true);
                }
            }
        }
    }

        /** Removes all waiting tasks from the task state table
         *  and from the sqs queue.
         */
    public void clearWaitingTask(String taskName) 
        throws SDBException, SQSException {
        synchronized(taskStateTable) {
            qops.deleteAllMessages();
            taskStateTable.deleteAllItems(true);
        }
    }


        // ----------- Single task ops --------------------------


        /** Removes a task from the task state table.  If the status
         *  of the state is 'waiting', this method also removes it
         *  from the sqs queue and removes its raw task object from
         *  S3.<p>
         *
         *  <b>Note/Warning:</b> if the specified task is 'waiting',
         *  this can be a relatively expensive operation.  SQS has
         *  no mechanism for retrieving a particular message.  So to
         *  get a message out of the queue, we need to retrieve <i>all</i>
         *  of contents and delete the target.
         * @param taskName
         */
    public void removeTask(String taskName) 
       throws SDBException, SQSException {
       if ( taskName == null ) return;
       synchronized(taskStateTable) {
           TaskState s = taskStateTable.getTaskState(taskName);
           if ( s == null ) return;
           if ( TaskState.Status.waiting.equals(s.status) ) {              
               Message m = getMessage(taskName);
               if ( m != null ) {
                   qops.deleteMessage(m);
               }
           }
           taskStateTable.deleteItem(taskName, true);
       }
    }

    private Message getMessage(String taskName) throws SQSException {
        List<Message> messages = qops.getAllMessages();
        for ( Message m : messages ) {
           TaskDescription des = new TaskDescription(m.getMessageBody());
           if ( taskName.equals(des.inputS3Key) ) return(m);
       }
        return(null);
    }

    public void requeueTask(String taskName) 
        throws SDBException, SQSException {

        if ( taskName == null ) return;
        TaskState s = taskStateTable.getTaskState(taskName);
        if ( s == null ) return;

        if ( s.taskDes == null ) { // If the des is missing, we can't run this,
                                   // so mark it an orphan and do not requeue.
            taskStateTable.recordUnrunnableTask(taskName, System.currentTimeMillis());
            return;
        }

            // Put the task des into the queue
        qops.queue.sendMessage(s.taskDes.toMessageString());

            // Update the task state.
        taskStateTable.recordRequeuedTask(s.taskDes,System.currentTimeMillis());
    }

    public void requeueTask(TaskState s)
        throws SDBException, SQSException {

        if ( s == null ) return;

        if ( s.taskDes == null ) { // If the des is missing, we can't run this,
                                   // so mark it an orphan and do not requeue.
            taskStateTable.recordUnrunnableTask(s.taskName,
                                                System.currentTimeMillis());
            return;
        }

            // Put the task des into the queue
        qops.queue.sendMessage(s.taskDes.toMessageString());

            // Update the task state.
        taskStateTable.recordRequeuedTask(s.taskDes,System.currentTimeMillis());
    }

    public void removeTaskFromQueue(TaskState s) throws SQSException {

        List<Message> messages = qops.getAllMessages();

        for ( Message m : messages ) {
            TaskDescription td = new TaskDescription(m.getMessageBody());
            if ( (td.inputS3Key != null) &&
                 td.inputS3Key.equals(s.taskName) ) { // Got it
                qops.deleteMessage(m);
                return;
            }
        }
    }

        // --------- Human interface ('Hi') single task ops ---------------

    public void dequeueTaskHi(final TaskState s) {
        if ( s == null ) return;
        if ( !GuiUtil.answerIsYes(new String[] {
                "Task name: ",
                "    " + s.taskName,
                "Status: " + s.status.toString(),
                "Submission time: " + Time.toTimeDate(s.submissionTime),
                "   ",
                "Remove this task from the input queue?",
                "   ",
                "(This will not affect its raw task Runnable object in S3.)"},
            "Re-queue task") )
            return;

       new TtoDequeueTask(s, "De-queue task");
    }

    public void deleteTaskHi(TaskState s) {
        if ( s == null ) return;

        if ( !GuiUtil.answerIsYes(new String[] {
                "Task name: ",
                "    " + s.taskName,
                "Status: " + s.status.toString(),
                "Submission time: " + Time.toTimeDate(s.submissionTime),
                "   ",
                "Delete this task?",
                "   ",
                "The task will be removed from the input queue AND",
                "it's raw Runnable object will be deleted from S3."},
            "Re-queue task") )
            return;

        new TtoDeleteTask(s,"Delete task");
    }

    public void clearTaskHi(TaskState s) {
        if ( s == null ) return;
        new TtoClearTask(s,"Clear task");
    }

    public void requeueTaskHi(TaskState s)
        throws SDBException, SQSException {

        if ( s == null ) return;

        if ( s.taskDes == null ) {
            GuiUtil.notice(new String[] {
                "The task you selected, ",
                "    " + s.taskName,
                "does not have a valid task description, to it",
                "cannot be re-queued." },
                "Cannot re-queue");
            return;
        }

        if ( !GuiUtil.answerIsYes(new String[] {
                "Task name: ",
                "    " + s.taskName,
                "Status: " + s.status.toString(),
                "Submission time: " + Time.toTimeDate(s.submissionTime),
                "   ",
                               "Put this task back in the input queue?" },
            "Re-queue task") )
            return;        

        new TtoRequeueTask(s,"Re-queue task");
    }

    abstract private class ThreadedTaskOp implements Runnable {

        TaskState s;
        String opDisplayName;

        ThreadedTaskOp(TaskState s,String opDisplayName) {
            this.s = s;
            this.opDisplayName = opDisplayName;
            (new Thread(this)).start();
        }

        public void run() {
            TaskInProgressWindow tip = null;
            try {
                tip = TaskInProgressWindow.run("RunJRun Monitor",
                                "In progress",
                                "<html>'" + opDisplayName + "'<br> is running.");
                doOp();
            }
            catch(Throwable x) {
                System.out.println(StringUtil.throwableToString(x));
            }
            finally { if ( tip != null ) tip.dispose(); }
        }

        abstract void doOp() throws Exception;
    }

    private class TtoRequeueTask extends ThreadedTaskOp {
        TtoRequeueTask(TaskState s,String opDisplayName) { super(s,opDisplayName); }
        void doOp() throws Exception {
                // Put the task des into the queue
            qops.queue.sendMessage(s.taskDes.toMessageString());

                // Update the task state.
            taskStateTable.recordRequeuedTask(s.taskDes,System.currentTimeMillis());
        }
    }

    private class TtoDequeueTask extends ThreadedTaskOp {
        TtoDequeueTask(TaskState s,String opDisplayName) { super(s,opDisplayName); }
        void doOp() throws Exception {
            removeTaskFromQueue(s);
        }
    }

    private class TtoClearTask extends ThreadedTaskOp {
        TtoClearTask(TaskState s,String opDisplayName) { super(s,opDisplayName); }
        void doOp() throws Exception {
            taskStateTable.deleteItem(s.taskName,true);
        }
    }

    private class TtoDeleteTask extends ThreadedTaskOp {
        TtoDeleteTask(TaskState s,String opDisplayName) { super(s,opDisplayName); }
        void doOp() throws Exception {
            sops.deleteObject(s.taskDes.inputBucketName,
                              s.taskDes.inputS3Key, true);
            taskStateTable.deleteItem(s.taskName,true);
        }
    }

        // ----------- Single node ops --------------------------

    public void clearNodeHi(NodeState s) {
        if ( s == null ) return;
        new TnoClearNode(s,"Clear node");
    }

    public void rebootInstanceHi(NodeState s) {
        if ( s == null ) return;

        if ( !GuiUtil.answerIsYes(new String[] {
            "Ec2 instance:",
            "ID=" + s.instanceID,
            "host=" + s.host,
            "node status=" + s.status,
            "instance state=" + s.instanceState,
            "    ",
            "Reboot it?"},
           "Reboot instance?") )
           return;        

        new TnoRebootInstance(s,"Reboot instance");
    }

    public void terminateInstanceHi(NodeState s) {
        if ( s == null ) return;
        if ( !GuiUtil.answerIsYes(new String[] {
            "Ec2 instance:",
            "ID=" + s.instanceID,
            "host=" + s.host,
            "node status=" + s.status,
            "instance state=" + s.instanceState,
            "    ",
            "Terminate it?"},
           "Termimate instance?") )
           return;

        new TnoTerminateInstance(s,"Reboot instance");
    }


    abstract private class ThreadedNodeOp implements Runnable {

        NodeState s;
        String opDisplayName;

        ThreadedNodeOp(NodeState s,String opDisplayName) {
            this.s = s;
            this.opDisplayName = opDisplayName;
            (new Thread(this)).start();
        }

        public void run() {
            TaskInProgressWindow tip = null;
            try {
                tip = TaskInProgressWindow.run("RunJRun Monitor",
                                "In progress",
                                "<html>'" + opDisplayName + "'<br> is running.");
                doOp();
            }
            catch(Throwable x) {
                System.out.println(StringUtil.throwableToString(x));
            }
            finally { if ( tip != null ) tip.dispose(); }
        }

        abstract void doOp() throws Exception;
    }

    private class TnoClearNode extends ThreadedNodeOp {
        TnoClearNode(NodeState s,String opDisplayName) { super(s,opDisplayName); }
        void doOp() throws Exception {
            nodeStateTable.deleteItem(s.host,true);
        }
    }

    private class TnoRebootInstance extends ThreadedNodeOp {
        TnoRebootInstance(NodeState s,String opDisplayName) { super(s,opDisplayName); }
        void doOp() throws Exception {
           if ( s.instanceID == null ) return;
           eops.rebootInstance(s.instanceID);
        }
    }

    private class TnoTerminateInstance extends ThreadedNodeOp {
        TnoTerminateInstance(NodeState s,String opDisplayName) { super(s,opDisplayName); }
        void doOp() throws Exception {
           if ( s.instanceID == null ) return;
           eops.terminateInstance(s.instanceID);
        }
    }
}
